package com.rocketmq.example;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

public class E06_EntiretyOrderMessage {

    private static final String TOPIC = "EntiretyOrderMessage";

    private static final String GROUP = "EntiretyOrderMessage";



    /**
     * 有序队列的原理就是将同一组需要顺序消费的消息放在一个消息队列里
     * 因为队列原本就是先进先出，所以就实现了顺序消费
     * ----
     * 全局顺序消费就是将消息全部发送到一个消息队列里
     * ----
     * 部分顺序消费就是将同一组消息发送到同一个队列里
     */


    /**
     * 使用 ConsumeOrderlyContext 保证消息消费的顺序性 
     * 这里发现消费有时会不及时，
     * 网上找资料应该是因为关闭消费者的时候没有执行 ConsumeMessageOrderlyService#shutdown  ，
     * 导致锁没有被主动释放，而是等待 60s 的锁过期时间后，新的实例才拿到队列锁开始消费
     * ----
     * 如果消费者一直存活，则新消息消费的时间是正常的
     * 
     * @param args
     */

    public static void main(String[] args) {
        consumer1();
        consumer2();
        consumer3();
    }




    public static void consumer1(){
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP);
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            consumer.subscribe(TOPIC, "*");

            // 设置为广播模式
            consumer.setMessageModel(MessageModel.CLUSTERING);

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerOrderly() {

                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    msgs.forEach(msg -> {
                        System.out.printf("消费消息:%s",
                                Thread.currentThread().getName() + ":" + new String(msg.getBody()) + "\n");
                    });
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
            System.err.println("消费者启动成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void consumer2(){
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP);
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            consumer.subscribe(TOPIC, "*");

            // 设置为广播模式
            consumer.setMessageModel(MessageModel.CLUSTERING);

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerOrderly() {

                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    msgs.forEach(msg -> {
                        System.out.printf("消费消息:%s",
                                Thread.currentThread().getName() + ":" + new String(msg.getBody()) + "\n");
                    });
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
            System.err.println("消费者启动成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void consumer3(){
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP);
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            consumer.subscribe(TOPIC, "*");

            // 设置为广播模式
            consumer.setMessageModel(MessageModel.CLUSTERING);

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerOrderly() {

                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    msgs.forEach(msg -> {
                        System.out.printf("消费消息:%s",
                                Thread.currentThread().getName() + ":" + new String(msg.getBody()) + "\n");
                    });
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
            System.err.println("消费者启动成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
